package it.unibo.contact.dspace;

import java.util.Hashtable;
import alice.tuprolog.*;
import it.unibo.is.interfaces.IMessage;
import it.unibo.is.interfaces.platforms.IMessageAndContext;
//import it.unibo.contact.platform.Consts;

public class Space extends SpaceSupport{
protected int maxNum = 1000;
protected static Hashtable<String,ExecThread> curExec ;
	public Space() throws Exception {
		super();
		curExec = new Hashtable<String,ExecThread>();
 	}

	@Override
	public void doJob() throws Exception  {
		//Executed in the thread Subject
		for( int i = 0; i<maxNum; i++ )
				receiveAndExec(i);
	}//doJob
	
	protected void receiveAndExec(int n) throws Exception{
//		showMsg("receiving ... " + n + "/"+ maxNum);
		IMessageAndContext requestM = hl_space_grant_coreCmd();
		IMessage m = requestM.getReceivedMessage();
  		showMsg("received "+ m );
		String cmd = m.msgContent();
		int reqN = Integer.parseInt( m.msgNum() );
		int seqN = m.getMsgSeqNum();
 		showMsg("received " + n + "/"+ maxNum + " from " + m.msgEmitter() + " reqN=" +  reqN +  " seqN=" + seqN + " --> " + m  );
 		
//		showMsg("received CMD="+ cmd + " LOCALMSGN=" + m.msgNum() + "/" + m.getMsgSeqNum() + " from " + m.msgEmitter() );
		if( cmd.contains("timeOutexception")){
			System.exit(1);
			return;
		}
		Term requestTerm = Term.createTerm(cmd);
 		if( requestTerm instanceof Struct ){			
// 			showMsg("				**** requestTerm=" + requestTerm );
 			String action = ""+((Struct)requestTerm).getArg(0);
 			if( curExec.get( m.msgEmitter()) == null  ){
 				new ExecThread(requestM, m.msgEmitter(), action, reqN, seqN, (Struct)requestTerm).start();
 			}else{//executor for some source already running
 				if( action.equals("out") )
 					/*
 					 * We must assure that two signals sent form the same source are
 					 * stored in the shared space according to the same order of emission  
 					 */
 					while( curExec.get(m.msgEmitter()+action) != null ){
 						System.out.println(curExec.get(m.msgEmitter()+action)
 								+" RECEIVE waits for the end of the EXEC THREAD still executing " + cmd + " for " + m.msgEmitter());
						Thread.sleep(100);
 					}
 					new ExecThread(requestM, m.msgEmitter(), action, reqN, seqN, (Struct)requestTerm).start();	
  			}
		}	
 	}
	
	protected class ExecThread extends Thread{
	protected IMessageAndContext requestM;
	protected Struct reqT;
	protected int reqN ;
	protected int seqN ;
	protected String emitter;
	protected String action;
		public ExecThread(IMessageAndContext requestM, String emitter, String action, int reqN, int seqN, Struct reqT ){
 			this.requestM = requestM;
 			this.emitter = emitter;
 			this.action = action;
 			this.reqN = reqN;
 			this.seqN = seqN; 
			this.reqT = reqT;
			curExec.put(emitter+action, this);
		}
		
		public void run(){
 			try {
 				showMsg("   ExecThread STARTS for "+reqT);
//				Thread.sleep(500);
 				String cmd = ""+reqT.getArg(0);
 				String msg = ""+reqT.getArg(1);
  				if( cmd.equals("out") ) 
 					curExec.put(emitter, this);
				execute(  reqT, cmd, msg );
				//Remove the executor from curExec
 				if( cmd.equals("out") ) 
					curExec.remove(emitter+action);
	 			showMsg("   ExecThread for " + emitter + " ENDS"  );
			} catch (Exception e) {
				showMsg("EXCEPTION "  + e.getMessage() );;
			}
 	 	}

		protected void execute( Struct reqT,String cmd,String msg ) throws Exception{
		IMessage answer;
		int n;
 		  try {
//			showMsg("		cmd="+cmd + " of " + msg + " / " + n);
			if( cmd.equals("out")){
	 			showMsg("   ExecThread for " + emitter + " out msg="+  msg   );
	 			
 				 core.out( msg );
// 				showMsg("	ExecThread sending answer done to out  " );
				String answerS = buildAnswer("done",""+ reqN );
 				requestM.replyToCaller( answerS );
				showMsg("   ExecThread for " + emitter + " sent answer done to   " + answerS   );
			}else if( cmd.equals("in")){
	 			showMsg("   ExecThread for " + emitter +" in msg="+  msg   );
				answer = core.in( msg );
				requestM.replyToCaller(""+answer);
				showMsg("   ExecThread for " + emitter + " sent answer to in = " + answer     );
			}else if( cmd.equals("rd")){ 
	 			showMsg("   ExecThread for " + emitter +" rd msg="+  msg   );
				int umsgNum = Integer.parseInt( ""+((Struct)reqT).getArg(2) );
				answer = core.rd( umsgNum,msg );
 				lastMsgNum = answer.getMsgSeqNum();
//				requestM.replyToCaller(answer.basicToString());
 				String answMsg = buildAnswer( answer.toString(), ""+ lastMsgNum );
				requestM.replyToCaller( answMsg ); //
				showMsg("   ExecThread for " + emitter +" sent answer to rd = " + answer  );
			}else if( cmd.equals("rdw")){
//				Term msgTerm = Term.createTerm(msg);
				int umsgNum = Integer.parseInt( ""+((Struct)reqT).getArg(2) );
//	 			showMsg("   ExecThread for " + emitter +" reqT="+ reqT );
	 			showMsg("   ExecThread for " + emitter +" rdw msg="+  msg  + " umsgNum=" + umsgNum );
 				answer = core.rdw( umsgNum,msg );
// 				IMessage moreRecentanswer = null;
// 				/*Find the most recent message*/
// 				lastMsgNum = answer.getMsgSeqNum();
//				do{
// 					moreRecentanswer=core.rd( lastMsgNum,msg );					
// 					showMsg("   	ExecThread  moreRecentanswer = " + moreRecentanswer);
// 					if( moreRecentanswer != null ){
// 						answer = moreRecentanswer;
// 						lastMsgNum = answer.getMsgSeqNum();
// 					}
// 				}
// 				while( moreRecentanswer != null ) ;
  				lastMsgNum = answer.getMsgSeqNum();
 				showMsg("			Last global time read for" + emitter + " =  " + lastMsgNum );
 				String answMsg = buildAnswer( answer.toString(), ""+ lastMsgNum );
// 				String answMsg = buildAnswer( answer.toString(), ""+ reqN );
				requestM.replyToCaller( answMsg ); 
				showMsg("   ExecThread for " + emitter +" sent answer to rdw = "  + answMsg  );
			}else if( cmd.equals("rdwMostRecent")){
//				Term msgTerm = Term.createTerm(msg);
				int umsgNum = Integer.parseInt( ""+((Struct)reqT).getArg(2) );
//	 			showMsg("   ExecThread for " + emitter +" reqT="+ reqT );
	 			showMsg("   ExecThread for " + emitter +" rdw msg="+  msg  + " umsgNum=" + umsgNum );
 				answer = core.rdwMostRecent( umsgNum,msg );
   				lastMsgNum = answer.getMsgSeqNum();
 				showMsg("			Last global time read for" + emitter + " =  " + lastMsgNum );
 				String answMsg = buildAnswer( answer.toString(), ""+ lastMsgNum );
// 				String answMsg = buildAnswer( answer.toString(), ""+ reqN );
				requestM.replyToCaller( answMsg ); 
				showMsg("   ExecThread for " + emitter +" sent answer to rdw = "  + answMsg  );
			}	
 		  } catch (Exception e) {
			  	showMsg("ERROR "  + e.getMessage() );
//				requestM.replyToCaller( ""+e.getMessage() );
		  }
		}
	}
	
	protected String buildAnswer( String msg, String msgNum ){
		return "dspaceAnswer("+msg+","+msgNum+")";
	}
	
 
}
